package com.wyx.springbootkafkaconsumerone.config;

import javax.annotation.Resource;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.protocol.types.Field.Str;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;

/**
 * @author wyx
 * @version V TODO
 * @Date 2024-10-16
 */
@Configuration
public class KafkaErrorHandlerConfig {

  private static final Logger log = LoggerFactory.getLogger(KafkaErrorHandlerConfig.class);

  @Resource
  private KafkaTemplate<String, String> kafkaTemplate;

  private String DLQ_TOPIC_SUFFIX = "-dlq";

  @Bean
  public ConsumerAwareListenerErrorHandler dlqErrorHandler() {
    return new ConsumerAwareListenerErrorHandler() {
      @Override
      public Object handleError(Message<?> message, ListenerExecutionFailedException e, Consumer<?, ?> consumer) {
        log.info("dlqErrorHandler receive : " + message.toString());
        MessageHeaders headers = message.getHeaders();
        String topic = (String) headers.get("kafka_receivedTopic");

        kafkaTemplate.send(topic + DLQ_TOPIC_SUFFIX, message.getPayload().toString());
        return null;
      }
    };
  }

  @Bean
  public ConsumerAwareListenerErrorHandler resetOffsetErrorHandler() {
    return new ConsumerAwareListenerErrorHandler() {
      @Override
      public Object handleError(Message<?> message, ListenerExecutionFailedException e, Consumer<?, ?> consumer) {
        log.info("consumerAwareErrorHandler receive : " + message.toString());
        MessageHeaders headers = message.getHeaders();
        Long offset = (Long) headers.get("kafka_offset");
        String topic = (String) headers.get("kafka_receivedTopic");
        int partition = 0;

        // 设置每个分区的偏移量
        TopicPartition topicPartition = new TopicPartition(topic, partition);
        log.info("重置客户端offset位置 topic {} partition {} {}", topic, partition, offset);
        consumer.seek(topicPartition, offset);
        return null;
      }
    };
  }
}
